/*
 * Copyright 2015-2020 the original author or authors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      https://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.springframework.cloud.bus;

import com.fasterxml.jackson.databind.ObjectMapper;
import org.springframework.boot.autoconfigure.AutoConfigureAfter;
import org.springframework.boot.autoconfigure.AutoConfigureBefore;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.autoconfigure.data.redis.RedisAutoConfiguration;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.cloud.autoconfigure.LifecycleMvcEndpointAutoConfiguration;
import org.springframework.cloud.bus.event.RemoteApplicationEvent;
import org.springframework.cloud.function.json.JacksonMapper;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.MessageListener;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.listener.PatternTopic;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.data.redis.listener.adapter.MessageListenerAdapter;
import org.springframework.messaging.converter.AbstractMessageConverter;
import org.springframework.messaging.support.GenericMessage;

/**
 * @author ahaschool
 */
@Configuration
@ConditionalOnBusEnabled
@ConditionalOnBean(StringRedisTemplate.class)
@EnableConfigurationProperties(BusProperties.class)
@AutoConfigureBefore({BusAutoConfiguration.class})
@AutoConfigureAfter({LifecycleMvcEndpointAutoConfiguration.class,
        PathServiceMatcherAutoConfiguration.class,
        RedisAutoConfiguration.class})
public class BusRedisAutoConfiguration {
    private final BusProperties properties;

    public BusRedisAutoConfiguration(BusProperties properties) {
        this.properties = properties;
    }

    @Bean
    @ConditionalOnMissingBean(BusBridge.class)
    public RedisBusBridge redisBusBridge(StringRedisTemplate redisTemplate,
                                         ObjectMapper objectMapper) {
        JacksonMapper jacksonMapper = new JacksonMapper(objectMapper);
        return new RedisBusBridge(redisTemplate, properties, jacksonMapper);
    }

    @Bean
    public RedisMessageListenerContainer redisMessageListenerContainer(RedisConnectionFactory connectionFactory,
                                                                       AbstractMessageConverter busJsonConverter,
                                                                       BusConsumer busConsumer) {
        RedisMessageListenerContainer container = new RedisMessageListenerContainer();
        container.setConnectionFactory(connectionFactory);
        MessageListener adapter = redisMessageListener(busJsonConverter, busConsumer);
        container.addMessageListener(adapter, new PatternTopic(properties.getDestination()));
        return container;
    }

    @Bean
    public MessageListener redisMessageListener(AbstractMessageConverter busJsonConverter,
                                                BusConsumer busConsumer) {
        return new MessageListenerAdapter(new RedisMessageListener(busJsonConverter, busConsumer));
    }


    public static class RedisMessageListener {
        private final AbstractMessageConverter busJsonConverter;
        private final BusConsumer busConsumer;

        public RedisMessageListener(AbstractMessageConverter busJsonConverter, BusConsumer busConsumer) {
            this.busJsonConverter = busJsonConverter;
            this.busConsumer = busConsumer;
        }

        public void handleMessage(String message) {
            Object targetValue = busJsonConverter.fromMessage(new GenericMessage<>(message), RemoteApplicationEvent.class);
            busConsumer.accept((RemoteApplicationEvent) targetValue);
        }
    }
}
